package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import com.intct.util.HBaseUtil;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.client.Put;

import java.util.Map;

/**
 * @author gufg
 * @since 2025-07-27 16:46
 */
public class HBaseSinkFunction extends RichSinkFunction<JSONObject> {
    private String namespace;
    private String tableName;
    private String familyName;
    private String rowKeyName;

    public HBaseSinkFunction(String namespace, String tableName, String familyName, String rowKeyName){
        this.namespace = namespace;
        this.tableName = tableName;
        this.familyName = familyName;
        this.rowKeyName = rowKeyName;
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // {"op":"r","data":{"name":"已结束","state":"3002"},"ts_ms":1753585266172,"db":"travel","table":"order_state"}
        // op类型
        String op = jsonObj.getString(Constant.MYSQL_CDC_JSON_KEY_OP);

        // 数据
        JSONObject dataJSONObj = jsonObj.getJSONObject("data");

        // rowKey
        String state = dataJSONObj.getString(rowKeyName);

        // 保存数据到HBase
        Put put = new Put(state.getBytes());
        for (Map.Entry<String, Object> entry : dataJSONObj.entrySet()) {
            put.addColumn(familyName.getBytes(), entry.getKey().getBytes(), entry.getValue().toString().getBytes());
        }
        // 调用HBase工具类
        HBaseUtil.put(namespace, tableName, put);
    }
}
